package com.mall.system.mq;

import com.alibaba.fastjson.JSON;
import com.mall.common.config.YamlPropertySourceFactory;
import com.mall.common.entity.gateway.GatewayLog;
import com.mall.common.mq.RocketMqUtil;
import com.mall.system.convert.GatewayLogConverter;
import com.mall.system.entity.SysLog;
import com.mall.system.service.SysLogService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.PropertySource;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

/**
 * 网关日志消费者服务
 */
@Slf4j
@Component
@PropertySource(value = "classpath:rocketmq.yml", factory = YamlPropertySourceFactory.class)
public class GatewayLogConsumerService {
    
    @Autowired
    private SysLogService sysLogService;
    
    @Value("${rocketmq.name-server:localhost:9876}")
    private String nameServer;
    
    @Value("${rocketmq.consumer.gateway-log-group:mall-gateway-log-consumer}")
    private String consumerGroup;
    
    @Value("${rocketmq.consumer.gateway-log-topic:mall_gateway_log_queue}")
    private String gatewayLogTopic;
    
    @Value("${rocketmq.consumer.gateway-log-tag:mall_gateway_log_tag}")
    private String gatewayLogTag;
    
    private RocketMqUtil rocketMqUtil;
    
    @PostConstruct
    public void init() {
        try {
            log.info("开始初始化网关日志消费者，NameServer地址: {}", nameServer);
            if (nameServer == null || nameServer.isEmpty()) {
                throw new IllegalStateException("RocketMQ NameServer地址未配置");
            }
            
            rocketMqUtil = new RocketMqUtil(nameServer, consumerGroup);
            MessageListenerConcurrently listener = rocketMqUtil.createMessageListener(this::handleGatewayLogMessage);
            rocketMqUtil.initConsumer(gatewayLogTopic, gatewayLogTag, listener);
            log.info("网关日志消费者初始化成功");
        } catch (MQClientException e) {
            log.error("初始化网关日志消费者失败: ", e);
        } catch (IllegalStateException e) {
            log.error("初始化网关日志消费者失败，配置错误: ", e);
        }
    }
    
    /**
     * 处理网关日志消息
     * 
     * @param message 消息内容
     */
    private void handleGatewayLogMessage(String message) {
        try {
            GatewayLog gatewayLog = JSON.parseObject(message, GatewayLog.class);
            SysLog sysLog = GatewayLogConverter.convertToSysLog(gatewayLog);
            boolean result = sysLogService.saveLog(sysLog);
            if (!result) {
                log.error("从消息队列保存网关日志失败: {}", sysLog.getId());
            } else {
                log.info("从消息队列保存网关日志成功: {}", sysLog.getId());
            }
        } catch (Exception e) {
            log.error("处理网关日志消息失败: ", e);
            // 可以根据业务需求决定是否重新入队或其他处理
        }
    }
    
    @PreDestroy
    public void destroy() {
        if (rocketMqUtil != null) {
            rocketMqUtil.shutdown();
        }
    }
}